package com.juneyao.log.logcenter.elasticsearch.async;

import com.juneyao.log.logcenter.elasticsearch.ElasticStorageService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.joda.time.DateTime;
import org.springframework.data.elasticsearch.annotations.DateFormat;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.core.query.IndexQuery;

import java.util.ArrayList;
import java.util.List;

/**
 * @author quanbiao
 * @date 2018-06-30
 * es 异步保存数据
 */
public class ElasticStorageAsync implements Runnable {

    /**
     * es 存储
     */
    private ElasticStorageService storageService;

    /**
     * kafka consumer 批量获取数据
     */
    private List<ConsumerRecord<String, String>> recordMeg;

    /**
     * es 索引名称前缀
     */
    private String indexNamePrefix;

    /**
     * @see Thread#run()
     */
    @Override
    public void run() {
        String dateStr = new DateTime().toString("yyyy-MM-dd");
        List<IndexQuery> indexQueryList = new ArrayList<>();
        recordMeg.forEach(consumerRecord -> {
            IndexQuery indexQuery = new IndexQuery();
            indexQuery.setIndexName(indexNamePrefix + dateStr);
            indexQuery.setType("log");
            indexQuery.setSource(consumerRecord.value());
            indexQueryList.add(indexQuery);
        });
        storageService.storageIndexSource(indexQueryList);
    }

    public ElasticStorageService getStorageService() {
        return storageService;
    }

    public void setStorageService(ElasticStorageService storageService) {
        this.storageService = storageService;
    }

    public List<ConsumerRecord<String, String>> getRecordMeg() {
        return recordMeg;
    }

    public void setRecordMeg(List<ConsumerRecord<String, String>> recordMeg) {
        this.recordMeg = recordMeg;
    }

    public String getIndexNamePrefix() {
        return indexNamePrefix;
    }

    public void setIndexNamePrefix(String indexNamePrefix) {
        this.indexNamePrefix = indexNamePrefix;
    }

    public ElasticStorageAsync(ElasticStorageService storageService, List<ConsumerRecord<String, String>> recordMeg, String indexNamePrefix) {
        this.storageService = storageService;
        this.recordMeg = recordMeg;
        this.indexNamePrefix = indexNamePrefix;
    }
}
